﻿using KafkaUtil.Consumers;
using KafkaUtil.Integration;
using KafkaUtil.Interface;
using KafkaUtil.Logger;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;

namespace KafkaUtil
{
    public delegate void Callback(RecieveResult recieveResult);
    public class Kafka
    {
        private IServiceCollection servicesProducer = null;
        private IServiceCollection servicesConsumer = null;
        private IServiceCollection servicesLogger = null;

        private IKafkaProducerFactory kafkaProducerFactory = null;
        private DefaultKafkaConsumerProvider consumerProvider = null;
        private ILoggerProvider loggerFactory = null;

        public KafkaProducerOptions producerOptions = null;
        public KafkaConsumerOptions consumerOptions = null;
        public KafkaLoggerOptions loggerOptions = null;

        public Kafka()
        {
            servicesProducer = new ServiceCollection();
            servicesConsumer = new ServiceCollection();
            servicesLogger = new ServiceCollection();
        }

        /*****************************获取配置生产者***************************************/
        /*
         * 获取配置生产者
         */
        public IClientProducer GetProducer()
        {
            if (kafkaProducerFactory != null)
            {
                return kafkaProducerFactory.CreateClientProducer();
            }
            else
            { 
                return GetProducer(this.producerOptions);
            }
        }

        /*
         * 获取配置生产者
         */
        public IClientProducer GetProducer(KafkaProducerOptions producerOptions)
        {
            if (producerOptions == null)
            {
                return null;
            }
            this.producerOptions = producerOptions;

            //AddKafkaProducer是添加Kafka发布者的相关配置，可以指定一个名称，使用时使用IKafkaProducerFactory接口注入
            #region Kafka
            servicesProducer.AddKafkaProducer(options =>
            {
                options.BootstrapServers = producerOptions.BootstrapServers;
                options.InitializeCount = producerOptions.InitializeCount;
                options.Key = producerOptions.Key;
                options.Topic = producerOptions.Topic;
            });
            #endregion

            kafkaProducerFactory = (IKafkaProducerFactory)servicesProducer.BuildServiceProvider().GetService<IKafkaProducerFactory>();
            return kafkaProducerFactory.CreateClientProducer();
        }
        /*****************************获取配置生产者***************************************/

        /*****************************添加配置消费者***************************************/
        /*
         * 添加配置消费者
         */
        public DefaultKafkaConsumerProvider GetConsumer()
        {
            if (consumerProvider != null)
            {
                return consumerProvider;
            }
            else
            { 
                return GetConsumer(this.consumerOptions);
            }
        }

        /*
         * 添加配置消费者
         */
        public DefaultKafkaConsumerProvider GetConsumer(KafkaConsumerOptions consumerOptions)
        {
            return GetConsumer(consumerOptions, null);
        }

        /*
         * 添加配置消费者
         */
        public DefaultKafkaConsumerProvider GetConsumer(KafkaConsumerOptions consumerOptions, Callback callback)
        {
            if (consumerOptions == null)
            {
                return null;
            }
            this.consumerOptions = consumerOptions;

            #region Kafka
            //BootstrapServers：Kafka集群地址，多个地址之间使用逗号分隔。
            //GroupId：消费者的Group，注意了，Kafka以Group的形式消费消息，一个消息只会被同一Group中的一个消费者消费，另外，一般的，同一Group中的消费者应该实现相同的逻辑
            //EnableAutoCommit：是否自动提交，如果设置成true，那么消费者接收到消息就相当于被消费了，我们可以设置成false，然后在我们处理完逻辑之后手动提交。
            //AutoOffsetReset：自动重置offset的行为，默认是Latest，这是kafka读取数据的策略，有三个可选值：Latest，Earliest，Error，个人推荐使用Earliest
            servicesConsumer.AddKafkaConsumer(options =>
            {
                options.BootstrapServers = consumerOptions.BootstrapServers;
                options.EnableAutoCommit = consumerOptions.EnableAutoCommit;//自动提交
                options.GroupId = consumerOptions.GroupId;
                options.Subscribers = consumerOptions.Subscribers;
            }).AddListener(result =>
            {
                if (callback != null)
                {
                    callback(result);
                    result.Commit();
                }
            });
            //}).AddListener<KafkaConsumerListener>();//实现IKafkaConsumerListener接口完成消费逻辑
            #endregion

            consumerProvider = (DefaultKafkaConsumerProvider)servicesConsumer.BuildServiceProvider().GetService<IKafkaConsumerProvider>();
            return consumerProvider;
        }
        /*****************************添加配置消费者***************************************/

        /*****************************添加配置日志记录***************************************/
        /*
         * 添加配置日志记录
         */
        public ILogger GetLogger()
        {
            if (loggerFactory != null)
            {
                return loggerFactory.CreateLogger(loggerOptions.Category);
            }
            else
            {
                return GetLogger(loggerOptions);
            }
        }

        /*
         * 添加配置日志记录
         */
        public ILogger GetLogger(KafkaLoggerOptions loggerOptions)
        {
            if (loggerOptions == null)
            {
                return null;
            }
            this.loggerOptions = loggerOptions;

            //AddKafkaProducer是添加Kafka发布者的相关配置，可以指定一个名称，使用时使用IKafkaProducerFactory接口注入
            #region Kafka
            servicesLogger.AddKafkaLogger(options =>
            {
                options.BootstrapServers = loggerOptions.BootstrapServers;
                options.Category = loggerOptions.Category;
                options.InitializeCount = loggerOptions.InitializeCount;
                options.Key = loggerOptions.Key;
                options.MinLevel = loggerOptions.MinLevel;
                options.Topic = loggerOptions.Topic;
                options.ApplicationName = loggerOptions.ApplicationName;
            });
            #endregion

            loggerFactory = (ILoggerProvider)servicesLogger.BuildServiceProvider().GetService<ILoggerProvider>();
            return loggerFactory.CreateLogger(loggerOptions.Category);
        }

        /*
         * 修改日志配置
         * 主要是用于修改LogLevel
         */
        public void SetLoggerOptions(KafkaLoggerOptions loggerOptions)
        {
            ((KafkaLoggerProvider)loggerFactory).SetLoggerOptions(loggerOptions);
        }
        /*****************************添加配置日志记录***************************************/
    }

    /*
     * 消费示例
     * 实现了IKafkaConsumerListener接口的类
     */
    public class KafkaConsumerListener : IKafkaConsumerListener
    {
        public Task ConsumeAsync(RecieveResult recieveResult)
        {
            Console.WriteLine("KafkaConsumerListener:" + recieveResult.Message);
            recieveResult.Commit();
            return Task.CompletedTask;
        }
    }
}
